package com.twoandone.tms.realtime.app.dwd;

import com.twoandone.tms.realtime.utils.KafkaUtil;
import com.twoandone.tms.realtime.utils.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdWayBillOrderCargo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_waybill_order_cargo"));
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));
        //tableEnv.executeSql("select * from `topic_db`").print();
        //TODO 运单明细表
        Table order_cargo = tableEnv.sqlQuery("select  \n" +
                "data['id'] id,\n" +
                "data['order_id'] order_id,\n" +
                "data['cargo_type'] cargo_type,\n" +
                "data['volume_length'] volume_length,\n" +
                "data['volume_width'] volume_width,\n" +
                "data['volume_height'] volume_height,\n" +
                "data['weight'] weight,\n" +
                "data['remark'] remark,\n" +
                "data['create_time'] create_time,\n" +
                "ts,\n" +
                "proc_time\n" +
                "from `topic_db`\n" +
                "where `table` = 'order_cargo' and \n" +
                "`type` = 'insert'");
        tableEnv.createTemporaryView("order_cargo", order_cargo);
        //tableEnv.executeSql("select * from order_cargo").print();
        //TODO 运单表
        Table order_info = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['order_no'] order_no,\n" +
                "data['status'] status,\n" +
                "data['collect_type'] collect_type,\n" +
                "data['user_id'] user_id,\n" +
                "data['receiver_complex_id'] receiver_complex_id,\n" +
                "data['receiver_province_id'] receiver_province_id,\n" +
                "data['receiver_city_id'] receiver_city_id,\n" +
                "data['receiver_district_id'] receiver_district_id,\n" +
                "data['receiver_address'] receiver_address,\n" +
                "data['receiver_name'] receiver_name,\n" +
                "data['receiver_phone'] receiver_phone,\n" +
                "data['receive_location'] receive_location,\n" +
                "data['sender_complex_id'] sender_complex_id,\n" +
                "data['sender_province_id'] sender_province_id,\n" +
                "data['sender_city_id'] sender_city_id,\n" +
                "data['sender_district_id'] sender_district_id,\n" +
                "data['sender_address'] sender_address,\n" +
                "data['sender_name'] sender_name,\n" +
                "data['sender_phone'] sender_phone,\n" +
                "data['send_location'] send_location,\n" +
                "data['payment_type'] payment_type,\n" +
                "data['cargo_num'] cargo_num,\n" +
                "data['amount'] amount,\n" +
                "data['estimate_arrive_time'] estimate_arrive_time,\n" +
                "data['distance'] distance,\n" +
                "ts,\n" +
                "proc_time\n" +
                "from `topic_db` \n" +
                "where `table` = 'order_info'\n" +
                "and `type`= 'insert'");
        tableEnv.createTemporaryView("order_info", order_info);
        //tableEnv.executeSql("select * from order_info").print();

        //TODO 获取字典表
        tableEnv.executeSql(MySqlUtil.getBaseDicLookUpDDL());
        tableEnv.executeSql("select * from base_dic").print();
        //TODO 将运单表与运单明细表进行关联
        Table joinTable = tableEnv.sqlQuery("select \n" +
                "oc.id,\n" +
                "oc.order_id,\n" +
                "oc.cargo_type,\n" +
                "oc.create_time,\n" +
                "oi.order_no,\n" +
                "oi.status,\n" +
                "oi.collect_type,\n" +
                "oi.receiver_complex_id,\n" +
                "oi.receiver_province_id,\n" +
                "oi.receiver_city_id,\n" +
                "oi.receiver_district_id,\n" +
                "oi.receiver_address,\n" +
                "oi.receiver_name,\n" +
                "oi.receiver_phone,\n" +
                "oi.receive_location,\n" +
                "oi.sender_complex_id,\n" +
                "oi.sender_province_id,\n" +
                "oi.sender_city_id,\n" +
                "oi.sender_district_id,\n" +
                "oi.sender_address,\n" +
                "oi.sender_name,\n" +
                "oi.sender_phone,\n" +
                "oi.send_location,\n" +
                "oi.payment_type,\n" +
                "oi.cargo_num,\n" +
                "oi.amount,\n" +
                "oi.estimate_arrive_time,\n" +
                "oi.distance,\n" +
                "ocdic.name as cargo_type_name,\n" +
                "oidic.name as status_name,\n" +
                "oiidic.name as collect_type_name,\n" +
                "oi.ts \n" +
                "from order_cargo oc \n" +
                "join\n" +
                "order_info oi\n" +
                "on oc.order_id = oi.id \n" +
                "join `base_dic` for system_time as of oc.proc_time as ocdic\n" +
                "on cast(oc.cargo_type as bigint) = ocdic.id \n" +
                "join `base_dic` for system_time as of oi.proc_time as oidic\n" +
                "on cast(oi.status as bigint) = oidic.id \n" +
                "join `base_dic` for system_time as of oi.proc_time as oiidic\n" +
                "on cast(oi.collect_type as bigint) = oiidic.id");
        tableEnv.createTemporaryView("join_table", joinTable);
        //tableEnv.executeSql("select * from join_table").print();
        //创建动态表与kafka主题进行映射
        tableEnv.executeSql("create table dwd_waybill_order_cargo (\n" +
                "id string,\n" +
                "order_id string,\n" +
                "cargo_type string,\n" +
                "create_time string,\n" +
                "order_no string,\n" +
                "status string,\n" +
                "collect_type string,\n" +
                "receiver_complex_id string,\n" +
                "receiver_province_id string,\n" +
                "receiver_city_id string,\n" +
                "receiver_district_id string,\n" +
                "receiver_address string,\n" +
                "receiver_name string,\n" +
                "receiver_phone string,\n" +
                "receive_location string,\n" +
                "sender_complex_id string,\n" +
                "sender_province_id string,\n" +
                "sender_city_id string,\n" +
                "sender_district_id string,\n" +
                "sender_address string,\n" +
                "sender_name string,\n" +
                "sender_phone string,\n" +
                "send_location string,\n" +
                "payment_type string,\n" +
                "cargo_num string,\n" +
                "amount string,\n" +
                "estimate_arrive_time string,\n" +
                "distance string,\n" +
                "cargo_type_name string,\n" +
                "status_name string,\n" +
                "collect_type_name string,\n" +
                "ts string,\n" +
                "primary key (id) not enforced\n" +
                ")" + KafkaUtil.getUpsertKafkaDDL("dwd_waybill_order_cargo"));

        tableEnv.executeSql("insert into dwd_waybill_order_cargo select * from join_table");
        /* env.execute();*/
    }
}
